package com.kool.kadmin.system.ws;

import com.kool.kadmin.common.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * webSocket 给前端推消息
 *
 * @author luyu
 */
@ServerEndpoint("/notify/{userId}/{sid}")
@Component
@Slf4j
public class WebSocketSession {
    private Session session;
    private Long userId = null;
    //接收sid
    private String sid = "";
    //当前连接数
    private static volatile AtomicInteger onlineCount = new AtomicInteger(0);
    //用来存放每个客户端对应的WebSocketSession对象。
    private static ConcurrentHashMap<String, WebSocketSession> webSocketMap = new ConcurrentHashMap<>();
    //存放每个用户的sid，针对一个用户多个会话场景
    private static ConcurrentHashMap<Long, Set<String>> sidMap = new ConcurrentHashMap<>();

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") Long userId, @PathParam("sid") String sid) {
        this.session = session;
        this.userId = userId;
        this.sid = sid;
        webSocketMap.put(sid, this);
        addOnlineCount();
        Set<String> sids = sidMap.getOrDefault(userId, new HashSet<>());
        sids.add(sid);
        sidMap.put(userId, sids);
        log.debug("有新窗口开始监听:" + sid + ",当前连接数为" + getOnlineCount());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        webSocketMap.remove(sid);
        subOnlineCount();
        Set<String> sids = sidMap.getOrDefault(userId, new HashSet<>());
        sids.remove(sid);
        log.debug("有一连接关闭！当前连接为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.debug("收到来自【" + userId + "】的信息:" + message);
    }

    /**
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        error(error);
    }

    public void error(Throwable error) {
        log.error("【{}】socket发生错误", sid);
        log.error(error.getMessage(), error);
    }

    /**
     * 实现服务器主动推送
     */
    public void sendMessage(String message) {
        try {
            if (this.session.isOpen() && StringUtil.isNotEmpty(message)) {
                this.session.getBasicRemote().sendText(message);
            }
        } catch (IOException e) {
            error(e);
        }
    }

    /**
     * 群发自定义消息
     */
    public static void sendInfo(String message, Long userId) {
        log.debug("推送消息到【{}】,推送内容: {}", userId == null ? "所有人" : userId, message);
        if (userId == null) {
            for (WebSocketSession item : webSocketMap.values()) {
                item.sendMessage(message);
            }
        } else {
            Set<String> sids = sidMap.get(userId);
            if (sids != null) {
                for (String sid : sids) {
                    webSocketMap.get(sid).sendMessage(message);
                }
            }
        }
    }

    public static int getOnlineCount() {
        return onlineCount.get();
    }

    public static void addOnlineCount() {
        WebSocketSession.onlineCount.addAndGet(1);
    }

    public static void subOnlineCount() {
        WebSocketSession.onlineCount.addAndGet(-1);
    }
}
